package test.verticle

import cn.hutool.core.util.RandomUtil
import com.gitee.ixtf.guice.Jguice
import com.gitee.ixtf.vertx.verticle.QueueConsumerCoroutineVerticle
import io.vertx.core.eventbus.Message
import io.vertx.kotlin.coroutines.CoroutineVerticle
import io.vertx.kotlin.coroutines.coAwait
import io.vertx.kotlin.coroutines.toReceiveChannel
import io.vertx.rabbitmq.RabbitMQClient
import io.vertx.rabbitmq.RabbitMQMessage
import io.vertx.rabbitmq.RabbitMQOptions
import java.util.concurrent.atomic.AtomicInteger
import kotlinx.coroutines.delay
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch

const val TEST_ADDRESS = "__TEST_ADDRESS__"
val COUNT = AtomicInteger()

object TestQueueVerticle : QueueConsumerCoroutineVerticle<String>(TEST_ADDRESS) {
  override suspend fun handleQueue(reply: Message<String>) {
    delay(RandomUtil.randomLong(1000, 3000))
    if (RandomUtil.randomBoolean()) {
      throw RuntimeException()
    }
    log.warn(reply.body())
  }
}

object TestMQVerticle : CoroutineVerticle() {
  private val options: RabbitMQOptions by Jguice
  private val client: RabbitMQClient by lazy { RabbitMQClient.create(vertx, options) }

  override suspend fun start() {
    val mqConsumer = client.basicConsumer("").coAwait()
    val channel = mqConsumer.toReceiveChannel(vertx)
    val launch = launch {
      for (delivery in channel) {
        if (isActive.not()) break
        val deliveryTag = delivery.envelope().deliveryTag
        try {
          handle(delivery)
          client.basicAck(deliveryTag, false).coAwait()
        } catch (e: Exception) {
          client.basicNack(deliveryTag, false, false).coAwait()
        }
      }
    }
  }

  private fun handle(delivery: RabbitMQMessage) {
    TODO("Not yet implemented")
  }
}
